package com.yenroc.ho.netty;

import com.yenroc.ho.netty.enums.GlobalTransactionType;
import com.yenroc.ho.netty.vo.TransactionVo;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;

/**
 * @author heyanpeng
 * @date 2021/7/27
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    private final Logger log = LoggerFactory.getLogger(NettyServerHandler.class);

    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    // 事务组中的出现了多少个事务列表<事务组Id, 事务的列表>
    private static Map<String, List<TransactionVo>> transactionMap = new HashMap<>();

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        channelGroup.add(ctx.channel());
        log.info("handlerAdded当前channel数量{}", channelGroup.size());
    }

    public void channelActive(ChannelHandlerContext ctx) {
    }

//    /**
//     * 创建事务组，并且添加保存事务
//     * @param ctx
//     * @param msg
//     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object obj) throws Exception {
        log.info("事务中心接收到事务请求的数据：" + obj);
        TkMessageVo msg = (TkMessageVo) obj;
        String groupId = msg.getGroupId();
        // 如果当前事务组不存在,则创建事务组，add 添加事务组,则需要创建事务
        if (transactionMap.get(groupId) == null) {
            log.info("创建全局=[{}]的事务组信息", groupId);
            transactionMap.put(groupId, new ArrayList<>());// 进入业务逻辑前创建事务
        }

        TransactionVo transactionVo = new TransactionVo();
        transactionVo.setChannel(ctx.channel());
        transactionVo.setTransactionId(msg.getTransactionId());
        // 表示开始执行的事务,
        if (msg.getTransactionType() == null) {
            transactionMap.get(groupId).add(transactionVo);
        } else {
            // 表示事务结束,将事务得状态附加上去
            transactionMap.get(groupId).stream()
                    .filter(t -> t.getTransactionId().equals(msg.getTransactionId()))
                    .forEach(t -> t.setType(msg.getTransactionType()));

        }
        // 创建的事务都收到的结果反馈
        if (transactionMap.get(groupId).stream().allMatch(t-> t.getType() != null)) {
            TkMessageVo result = new TkMessageVo();
            result.setGroupId(groupId);
            // 其中有一个为rollback,则全部回滚
            if (transactionMap.get(groupId).stream().anyMatch(t-> t.getType() == GlobalTransactionType.rollback)) {
                result.setTransactionType(GlobalTransactionType.rollback);
            } else {
                result.setTransactionType(GlobalTransactionType.commit);
            }
            transactionMap.get(groupId).forEach(x->{
                result.setTransactionId(x.getTransactionId());
                sendResult(result, x.getChannel());
            });
            transactionMap.remove(groupId);
        }
    }

    /**
     * @param ctx   当前的应用上下文
     * @param cause Throwable是异常和Error的顶级接口,此处就是异常
     * @throws Exception
     */
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("NettyServerHandler 发生异常", cause);
    }

    private void sendResult(TkMessageVo message, Channel channel) {
        log.info("发送给[{}]数据：{}",channel.remoteAddress().toString(), message);
        channel.writeAndFlush(message);
    }
}
